package cn.zk.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.util.Objects;

/**
 * ClassName: CuratorThird
 * Desc: 事件监听 Cache
 * Curator 引入Cache 来实现对ZK服务端事件的监听。
 * Date： 2016/12/6
 * Created：shaom
 */
public class CuratorThird {

    public static final String connectString = "master:2181,slave01:2181,slave02:2181";


    /**
     * 节点监听
     * NodeCache:
     * 1、可以监听数据节点的内容变化
     * 2、监听指定节点是否存在。原本不存在，在节点创建后出发监听回调
     * 3、删除节点，也会触发
     *
     * @throws Exception
     */
    void testNodeCache() throws Exception {
        final String path = "/NodeCache";
        final String changeValue = "changeV";
        final String initValue = "initV";

        // 启动客户端
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();

        // 新建节点监听
        final NodeCache cache = new NodeCache(client, path, false);

        // 默认是false;
        // true:NodeCache在第一次启动的时候就会立刻从ZK上读取对应节点的数据内容，并保存在Cache中
        cache.start(true);

        // 事件处理的回调接口(数据节点的内容发生变化的时候，就会回调)
        cache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("~~~~~~~~~~触发监听~~~~~~~~~~");
                if (!Objects.isNull(cache.getCurrentData())) {
                    System.out.println("new data : " + new String(cache.getCurrentData().getData()));
                }
            }
        });

        // 创建节点(触动回调)
        client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath(path, initValue.getBytes());

        // 修改节点的值(触动回调)
        client.setData().forPath(path, changeValue.getBytes());
        Thread.sleep(2000);

        // 递归删除节点(触动回调)
        client.delete().forPath(path);
//        client.delete().deletingChildrenIfNeeded().forPath(path);
        Thread.sleep(Integer.MAX_VALUE);

    }


    /**
     * 子节点监听
     * PathChildrenCache
     * 二级节点不会触发
     *
     * @throws Exception
     */
    void testChildNodeCache() throws Exception {
        final String path = "/PathChildrenCache";
        final String changeValue = "changeV";
        final String initValue = "initV";

        // 启动客户端
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();

        // cacheData-是否把节点内容缓存起来。true-节点变更，同时获取内容一并缓存
        PathChildrenCache cache = new PathChildrenCache(client, path, true);
        cache.start();

        // 事件处理的回调接口
        cache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        System.out.println("CHILD_ADDED: ");
                        break;
                    case CHILD_UPDATED:
                        System.out.println("CHILD_UPDATED: ");
                        break;
                    case CHILD_REMOVED:
                        System.out.println("CHILD_REMOVED: ");
                        break;
                    case CONNECTION_SUSPENDED:
                        System.out.println("CONNECTION_SUSPENDED: ");
                        break;
                    case CONNECTION_RECONNECTED:
                        System.out.println("CONNECTION_RECONNECTED: ");
                        break;
                    case CONNECTION_LOST:
                        System.out.println("CONNECTION_LOST: ");
                        break;
                    case INITIALIZED:
                        System.out.println("INITIALIZED: ");
                        break;
                    default:
                        System.out.println("default : ");
                        break;
                }
            }
        });

        // 新建父节点
        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        Thread.sleep(1000);

        // 新建子节点
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        Thread.sleep(1000);

        // 修改子节点的数据
        client.setData().forPath(path + "/c1", changeValue.getBytes());
        Thread.sleep(1000);

        // 新建二级子节点
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path + "/c1/c2");
        Thread.sleep(1000);
        // 删除二级子节点
        client.delete().forPath(path + "/c1/c2");
        Thread.sleep(1000);

        // 删除子节点
        client.delete().forPath((path + "/c1"));
        Thread.sleep(1000);

        client.delete().forPath(path);
        Thread.sleep(1000);

    }

    /**
     * 支持所有节点监听
     *
     * @throws Exception
     */
    void testTreeCache() throws Exception {
        final String path = "/PathChildrenCache";
        final String changeValue = "changeV";
        final String initValue = "initV";

        // 启动客户端
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();

        // cacheData-是否把节点内容缓存起来。true-节点变更，同时获取内容一并缓存
        TreeCache cache = new TreeCache(client, path);
        cache.start();

        // 事件处理的回调接口
        cache.getListenable().addListener(new TreeCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case NODE_ADDED:
                        System.out.println("NODE_ADDED: " + event.getData());
                        break;
                    case NODE_UPDATED:
                        System.out.println("NODE_UPDATED: " + event.getData());
                        break;
                    case NODE_REMOVED:
                        System.out.println("NODE_REMOVED: " + event.getData());
                        break;
                    case CONNECTION_SUSPENDED:
                        System.out.println("CONNECTION_SUSPENDED: " + event.getData());
                        break;
                    case CONNECTION_RECONNECTED:
                        System.out.println("CONNECTION_RECONNECTED: " + event.getData());
                        break;
                    case CONNECTION_LOST:
                        System.out.println("CONNECTION_LOST: " + event.getData());
                        break;
                    case INITIALIZED:
                        System.out.println("INITIALIZED: " + event.getData());
                        break;
                    default:
                        System.out.println("default : " + event.getData());
                        break;
                }
            }
        });

        // 新建父节点
        client.create().withMode(CreateMode.PERSISTENT).forPath(path);
        Thread.sleep(1000);

        // 新建子节点
        client.create().withMode(CreateMode.PERSISTENT).forPath(path + "/c1");
        Thread.sleep(1000);

        // 修改子节点的数据
        client.setData().forPath(path + "/c1", changeValue.getBytes());
        Thread.sleep(1000);

        // 新建二级子节点
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path + "/c1/c2");
        Thread.sleep(1000);
        // 删除二级子节点
        client.delete().forPath(path + "/c1/c2");
        Thread.sleep(1000);

        // 删除子节点
        client.delete().forPath((path + "/c1"));
        Thread.sleep(1000);

        client.delete().forPath(path);
        Thread.sleep(1000);

//        INITIALIZED:
//        NODE_ADDED:
//        NODE_ADDED:
//        NODE_UPDATED:
//        NODE_ADDED:
//        NODE_REMOVED:
//        NODE_REMOVED:
//        NODE_REMOVED:
    }


    void deleteNode(String node) throws Exception {
        // 启动客户端
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();

        client.delete().deletingChildrenIfNeeded().forPath(node);

    }
}
